package com.jiayuan.cn.energy.test.one;

import com.jiayuan.cn.energy.test.ok.MqttConfig;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.TableSchema;
import org.apache.flink.table.sources.StreamTableSource;
import org.eclipse.paho.client.mqttv3.*;

import java.util.Arrays;

import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

public class MQTTTableSource implements StreamTableSource<String> {

    private final DeserializationSchema<String> deserializationSchema;
    static private final String topic;
    static private final String brokerUrl;
    static private final String username;
    static private final String password;
    static private final String clientID;
    static final MqttConfig mqttConfigBean;
    static final String[] msgtopic;
    static final int[] qos;

    static {
        topic = "/v1/report/+/model";
//        brokerUrl = "tcp://192.168.110.88:1883";
        brokerUrl = "tcp://127.0.0.1:1883";
        username = "s3";
        password = "111";
        clientID = "S916";
        mqttConfigBean = new MqttConfig(username, password, brokerUrl, clientID, topic);
        msgtopic = mqttConfigBean.getMsgTopic();
        //订阅消息
        qos = new int[msgtopic.length];
        for (int i = 0; i < msgtopic.length; i++) {
            qos[i] = 0;
        }
    }
    public MQTTTableSource() {
        this(null);
    }
    public MQTTTableSource(DeserializationSchema<String> deserializationSchema) {
        this.deserializationSchema = deserializationSchema;
    }
//    public MQTTTableSource(String clientID, String topic, String brokerUrl, String username, String password, DeserializationSchema<String> deserializationSchema) {
//        this.deserializationSchema = deserializationSchema;
//        this.topic = topic;
//        this.brokerUrl = brokerUrl;
//        this.username = username;
//        this.password = password;
//        mqttConfigBean = new MqttConfig(username, password, brokerUrl, clientID, topic);
//        msgtopic = mqttConfigBean.getMsgTopic();
//        //订阅消息
//        qos = new int[msgtopic.length];
//        for (int i = 0; i < msgtopic.length; i++) {
//            qos[i] = 0;
//        }
//    }

    @Override
    public TableSchema getTableSchema() {
        return new TableSchema(
                Arrays.asList("data").toArray(new String[0]),
                Arrays.asList(DataTypes.STRING()).toArray(new TypeInformation<?>[0])
        );
    }

    @Override
    public DataStream<String> getDataStream(StreamExecutionEnvironment execEnv) {
        return execEnv.addSource(new MQTTSourceFunction(topic, brokerUrl, deserializationSchema));
    }

    private static class MQTTSourceFunction implements SourceFunction<String> {

        private final String topic;
        private final String brokerUrl;
        private final DeserializationSchema<String> deserializationSchema;
        private MqttClient mqttClient;

        public MQTTSourceFunction(String topic, String brokerUrl, DeserializationSchema<String> deserializationSchema) {
            this.topic = topic;
            this.brokerUrl = brokerUrl;
            this.deserializationSchema = deserializationSchema;
        }

        @Override
        public void run(SourceContext<String> ctx) throws Exception {
//            mqttClient = new MqttClient(brokerUrl, MqttClient.generateClientId(),username,password);
//            MqttConnectOptions options = new MqttConnectOptions();
//            options.setAutomaticReconnect(true);
//            options.setCleanSession(true);
//            mqttClient.connect(options);
//            mqttClient.subscribe(topic);

            //连接mqtt服务器
            mqttClient = new MqttClient(mqttConfigBean.getHostUrl(), mqttConfigBean.getClientId(), new MemoryPersistence());
            MqttConnectOptions options = new MqttConnectOptions();
            options.setCleanSession(false);
            options.setUserName(mqttConfigBean.getUsername());
            options.setPassword(mqttConfigBean.getPassword().toCharArray());
            options.setCleanSession(false);   //是否清除session
            // 设置超时时间
            options.setConnectionTimeout(30);
            // 设置会话心跳时间
            options.setKeepAliveInterval(20);

            mqttClient.setCallback(new MqttCallback() {
                @Override
                public void connectionLost(Throwable cause) {
                    // Handle connection loss
                    System.out.println("MQTT连接断开，发起重连");
                    while (true) {
                        try {
                            Thread.sleep(1000);
                            mqttClient.connect(options);
                            //订阅消息
                            mqttClient.subscribe(msgtopic, qos);
                            System.out.println("MQTT重新连接成功:" + mqttClient);
                            break;
                        } catch (Exception e) {
                            e.printStackTrace();
                            continue;
                        }
                    }
                }

                @Override
                public void messageArrived(String topic, MqttMessage message) throws Exception {
                    System.out.println("topic:" + topic + "; message:" + message);
                    //订阅消息字符
//                    String msg = new String(message.getPayload());
//                    byte[] bymsg = getBytesFromObject(msg);
//                    System.out.println("topic:" + topic);
//                    queue.put(msg);
                    String data = new String(message.getPayload());
                    ctx.collect(data);
//                    ctx.collect(deserializationSchema.deserialize(data.getBytes()));
                }

                @Override
                public void deliveryComplete(IMqttDeliveryToken token) {
                    // Handle delivery completion
                }
            });
        }

        @Override
        public void cancel() {
            try {
                mqttClient.disconnect();
            } catch (MqttException e) {
                e.printStackTrace();
            }
        }
    }
}